package kafkax

import (
	"errors"
	"github.com/Shopify/sarama"
	"time"
)

func NewProducer(addr []string) (*Producer, error) {
	if len(addr) == 0 {
		return nil, errors.New("addr required")
	}

	c := sarama.NewConfig()
	c.Producer.RequiredAcks = sarama.WaitForAll
	c.Producer.Return.Successes = true

	p, err := sarama.NewSyncProducer(addr, c)
	if err != nil {
		return nil, err
	}

	return &Producer{
		addr:     addr,
		producer: p,
	}, nil
}

type Producer struct {
	config   *sarama.Config
	producer sarama.SyncProducer // 生产者
	addr     []string
}

func (k *Producer) Close() {
	defer func() {
		if k.producer != nil {
			k.producer.Close()
		}
	}()
}

// SendMsg 生产消息
func (k *Producer) SendMsg(topic string, str string) (partition int32, offset int64, err error) {
	msg := &sarama.ProducerMessage{}
	msg.Topic = topic
	msg.Value = sarama.ByteEncoder(str)
	msg.Offset = 0
	msg.Timestamp = time.Now()
	return k.producer.SendMessage(msg)
}
